package com.lsp.test.kafka.transaction;

import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.KafkaException;

import java.util.Properties;
import java.util.UUID;

public class KafkaProducer {
    public static void main(String[] args) {
        Properties pros=new Properties();
//        pros.put("bootstrap.servers","192.168.8.144:9092,192.168.8.145:9092,192.168.8.146:9092");
        pros.put("bootstrap.servers","47.107.111.118:9092");
        pros.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
        pros.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
        // 0 发出去就确认 | 1 leader 落盘就确认| all(-1) 所有Follower同步完才确认
        //Must set acks to all in order to use the idempotent producer. Otherwise we cannot guarantee idempotence.
        //事务必须设置为ALL
        pros.put("acks","all");
        // 异常自动重试次数
        pros.put("retries",3);
        // 多少条数据发送一次，默认16K
        pros.put("batch.size",16384);
        // 批量发送的等待时间
        pros.put("linger.ms",5);
        // 客户端缓冲区大小，默认32M，满了也会触发消息发送
        pros.put("buffer.memory",33554432);
        // 获取元数据时生产者的阻塞时间，超时后抛出异常
        pros.put("max.block.ms",3000);

        //也就是说没有办法通过直接配置开启,也就无法使用事物,当然可以选择手动配置Kafka
        // 消息幂等性
        pros.put("enable.idempotence",true);

        // 事务ID，唯一
        pros.put("transactional.id", UUID.randomUUID().toString());

        // 创建Sender线程
        Producer<String,String> producer = new org.apache.kafka.clients.producer.KafkaProducer<String,String>(pros);


        // 初始化事务
        producer.initTransactions();
        try {
            // 事务开始
            producer.beginTransaction();
            producer.send(new ProducerRecord<String, String>("test", "1", "1"));
//                        Integer i = 1/0;
            producer.send(new ProducerRecord<String, String>("test", "2", "2"));
            // 提交事务
            producer.commitTransaction();
        }catch (KafkaException e){
            // 中止事务
            producer.abortTransaction();
        }
        producer.close();
    }
}
